import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;


import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.SequenceFile.CompressionType;

  public class rv {

    public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, LongWritable, LongWritable> {
      static Random r = new Random();

      public void map(LongWritable key, Text value, OutputCollector<LongWritable, LongWritable> output, Reporter reporter) throws IOException {
        long nSample = Long.parseLong(value.toString());
        int hold;

        for( int i=0; i<nSample; i=i+1){
          //generate a random number
          hold = r.nextInt(10);
          //store into collector
          output.collect(new LongWritable(hold), new LongWritable(1));
        }//end for
      }//end map
    }

    public static class Reduce extends MapReduceBase implements Reducer<LongWritable, LongWritable, LongWritable, LongWritable> {
      public void reduce(LongWritable key, Iterator<LongWritable> values, OutputCollector<LongWritable, LongWritable> output, Reporter reporter) throws IOException {
         int sum = 0;
         while (values.hasNext()) {
          sum += values.next().get();  //count all values that have the same keys
         }
        //store into sequence file
        output.collect(key, new LongWritable(sum));
      }//end reduce
    }

    public static void main(String[] args) throws Exception {
      JobConf conf = new JobConf(rv.class);
      conf.setJobName("rv");

      conf.setOutputKeyClass(LongWritable.class);
      conf.setOutputValueClass(LongWritable.class);

      conf.setMapperClass(Map.class);
      conf.setReducerClass(Reduce.class);

      //set numbers of reducers
      conf.setNumReduceTasks(1);

      conf.setInputFormat(TextInputFormat.class);        // text file
      //conf.setOutputFormat(SequenceFileOutputFormat.class);  // output gfs format, unreadable
      conf.setOutputFormat(TextOutputFormat.class);  //sequence file

      FileInputFormat.setInputPaths(conf, new Path(args[0]));
      FileOutputFormat.setOutputPath(conf, new Path(args[1]));

      //exec JOB
      JobClient.runJob(conf);

      /**
      //get Data from SequenceFileOutputFormat
      FileSystem fileSys = FileSystem.get(conf);
      SequenceFile.Reader reader = new SequenceFile.Reader(fileSys, new Path(args[1],"part-00000"), conf);

      LongWritable key = new LongWritable();
      LongWritable value = new LongWritable();

      System.out.println(args[0]+ "->" + args[1]);
      while(reader.next(key, value)){
        System.out.println(key.get()+ "->" + value.get());
      }//end while
      reader.close();
      //*/
    }
  }
